package com.ucode.dcse;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.annotation.Resource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.ucode.dcse.handler.TableDataHandlerContext;
import com.ucode.springboot.starter.canal.annotation.CanalEventListener;
import com.ucode.springboot.starter.canal.annotation.ddl.AlertTableListenPoint;
import com.ucode.springboot.starter.canal.annotation.ddl.CreateIndexListenPoint;
import com.ucode.springboot.starter.canal.annotation.ddl.CreateTableListenPoint;
import com.ucode.springboot.starter.canal.annotation.ddl.DropTableListenPoint;
import com.ucode.springboot.starter.canal.annotation.dml.DeleteListenPoint;
import com.ucode.springboot.starter.canal.annotation.dml.InsertListenPoint;
import com.ucode.springboot.starter.canal.annotation.dml.UpdateListenPoint;
import com.ucode.springboot.starter.canal.client.CanalMsg;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;

/**
 *  Canal 监听器(注解方式实现的监听器)
 * @author: liliang
 * @date: 2020年2月28日 下午3:12:21
 */
@CanalEventListener
public class CanalAnnoEventListener {
    private static Log log = LogFactory.get();
    
    // disruptor异步队列
    @Resource(name="canalSyncEsDisruptorProducer")
    private CanalSyncEsDisruptorProducer disruptorProducer;
    
    @Autowired
    private TableDataHandlerContext tableDataHandlerContext;
    
    /**
     * 新增数据监听器
     * @date 2020年2月28日
     * @param canalMsg
     * @param rowChange
     */
    @InsertListenPoint(schema={"ucode_crm"})
    public void onEventInsertData(CanalMsg canalMsg, CanalEntry.RowChange rowChange) {
        
        TableCanalData tableCanalData = new TableCanalData();
        tableCanalData.setDestination(canalMsg.getDestination());
        tableCanalData.setEventType(CanalEntry.EventType.INSERT);
        tableCanalData.setSchemaName(canalMsg.getSchemaName());
        tableCanalData.setTableName(canalMsg.getTableName());
        JSONArray afterDatas = new JSONArray();
        
        Map<String, String> mapping = tableDataHandlerContext.getMapping(canalMsg.getSchemaName(), canalMsg.getTableName());
        
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        /**
         * RowData --具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理
         * insert 只有after数据，delete只有 before数据，update会有after和before数据
         */
        for (CanalEntry.RowData rowData : rowDatasList) {
            
            //insert只有after columns, delete只有before columns，而update则会有before / after columns数据.
            //变更后的字段数据
            List<CanalEntry.Column> afterColumns = rowData.getAfterColumnsList();
            
            String sql = "use " + canalMsg.getSchemaName() + ";\n";
            StringBuffer colums = new StringBuffer();
            StringBuffer values = new StringBuffer();
            
            JSONObject afterData = new JSONObject();
            afterColumns.forEach((c) -> {
                colums.append(c.getName() + ",");
                values.append("'" + c.getValue() + "',");
                
                String property = null;
                if(mapping == null || StrUtil.isBlank(property = mapping.get(c.getName()))){
                    property = c.getName();
                }
                afterData.put(property, c.getValue());
                
            });
            
            
            sql += "INSERT INTO " + canalMsg.getTableName() + "(" + colums.substring(0, colums.length() - 1) + ") VALUES(" + values.substring(0, values.length() - 1) + ");";
            
            log.info(sql);
            
//            sqls.add(sql);
            afterDatas.add(afterData);
        }
        tableCanalData.setAfterDatas(afterDatas);
        disruptorProducer.send(Arrays.asList(tableCanalData));
        
    }
    
    /**
     * 更新数据监听器
     * @date 2020年2月28日
     * @param canalMsg
     * @param rowChange
     */
    @UpdateListenPoint(schema={"ucode_crm"})
    public void onEventUpdateData(CanalMsg canalMsg, CanalEntry.RowChange rowChange) {
        
        TableCanalData tableCanalData = new TableCanalData();
        tableCanalData.setDestination(canalMsg.getDestination());
        tableCanalData.setEventType(CanalEntry.EventType.UPDATE);
        tableCanalData.setSchemaName(canalMsg.getSchemaName());
        tableCanalData.setTableName(canalMsg.getTableName());
        JSONArray afterDatas = new JSONArray();
        
        JSONArray beforeDatas = new JSONArray();
        
        Map<String, String> mapping = tableDataHandlerContext.getMapping(canalMsg.getSchemaName(), canalMsg.getTableName());
        
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        /**
         * RowData --具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理
         * insert 只有after数据，delete只有 before数据，update会有after和before数据
         */
        for (CanalEntry.RowData rowData : rowDatasList) {
            
            //变更前的字段数据
            List<CanalEntry.Column> beforeColumns = rowData.getBeforeColumnsList();
            //变更后的字段数据
            List<CanalEntry.Column> afterColumns = rowData.getAfterColumnsList();
            
            JSONObject beforeData = new JSONObject();
            beforeColumns.forEach((c) -> {
                String property = null;
                if(mapping == null || StrUtil.isBlank(property = mapping.get(c.getName()))){
                    property = c.getName();
                }
                beforeData.put(property, c.getValue());
                
            });
            beforeDatas.add(beforeData);
            
            JSONObject afterData = new JSONObject();
            
            String sql = "use " + canalMsg.getSchemaName() + ";\n";
            StringBuffer updates = new StringBuffer();
            StringBuffer conditions = new StringBuffer();
            
            afterColumns.forEach((c) -> {
                if (c.getIsKey()) {
                    conditions.append(c.getName() + "='" + c.getValue() + "'");
                } else {
                    updates.append(c.getName() + "='" + c.getValue() + "',");
                }
                
                String property = null;
                if(mapping == null || StrUtil.isBlank(property = mapping.get(c.getName()))){
                    property = c.getName();
                }
                afterData.put(property, c.getValue());
                
            });
            
            sql += "UPDATE " + canalMsg.getTableName() + " SET " + updates.substring(0, updates.length() - 1) + " WHERE " + conditions;
            
            log.info(sql);
            
//            sqls.add(sql);
            afterDatas.add(afterData);
        }
        
//        tableCanalData.setSqls(sqls);
        tableCanalData.setAfterDatas(afterDatas);
        tableCanalData.setBeforeDatas(beforeDatas);
        disruptorProducer.send(Arrays.asList(tableCanalData));
    }
    
    
    @DeleteListenPoint(schema={"ucode_crm"})
    public void onEventDeleteData(CanalMsg canalMsg,CanalEntry.RowChange rowChange) {
        
        TableCanalData tableCanalData = new TableCanalData();
        tableCanalData.setDestination(canalMsg.getDestination());
        tableCanalData.setEventType(CanalEntry.EventType.DELETE);
        tableCanalData.setSchemaName(canalMsg.getSchemaName());
        tableCanalData.setTableName(canalMsg.getTableName());
        
        JSONArray beforeDatas = new JSONArray();
        
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        /**
         * RowData --具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理
         * insert 只有after数据，delete只有 before数据，update会有after和before数据
         */
        for (CanalEntry.RowData rowData : rowDatasList) {
            //变更前的字段数据
            List<CanalEntry.Column> beforeColumns = rowData.getBeforeColumnsList();
            
            if (!CollectionUtils.isEmpty(beforeColumns)) {
                String sql = "use " + canalMsg.getSchemaName() + ";\n";
                
                sql += "DELETE FROM " + canalMsg.getTableName() + " WHERE ";
                StringBuffer idKey = new StringBuffer();
                StringBuffer idValue = new StringBuffer();
                for (CanalEntry.Column c : rowData.getBeforeColumnsList()) {
                    if (c.getIsKey()) {
                        idKey.append(c.getName());
                        idValue.append(c.getValue());
                        
                        beforeDatas.add(c.getValue());
                        break;
                    }
                }
                
                sql += idKey + " =" + idValue + ";";
                log.info(sql);
            }
            
            tableCanalData.setBeforeDatas(beforeDatas);
        }
        
        disruptorProducer.send(Arrays.asList(tableCanalData));
    }
    
    @CreateTableListenPoint(schema={"ucode_crm"})
    public void onEventCreateTable(CanalMsg canalMsg,CanalEntry.RowChange rowChange) {
        log.info("use " + rowChange.getDdlSchemaName() + ";\n" + rowChange.getSql());
        
    }
    
    @DropTableListenPoint(schema={"ucode_crm"})
    public void onEventDropTable(CanalMsg canalMsg,CanalEntry.RowChange rowChange) {
        log.info("use " + rowChange.getDdlSchemaName() + ";\n" + rowChange.getSql());
    }
    
    @AlertTableListenPoint(schema={"ucode_crm"})
    public void onEventAlertTable(CanalMsg canalMsg,CanalEntry.RowChange rowChange) {
        log.info("use " + rowChange.getDdlSchemaName() + ";\n" + rowChange.getSql());
    }
    
    @CreateIndexListenPoint(schema={"ucode_crm"})
    public void onEventCreateIndex(CanalMsg canalMsg,CanalEntry.RowChange rowChange){
        log.info("use " + canalMsg.getSchemaName()+ ";\n" + rowChange.getSql());
    }
    
}
